package cn.itcast.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test


class SparkWordCount {
  def main(args: Array[String]): Unit = {

    // 1、创建SparkContext
    val conf = new SparkConf().setMaster("local[2]").setAppName("SparkWordCount")
    val sc = new SparkContext(conf)

    // 2、加载文件
    //    1、准备文件
    //    2、读取文件

    // RDD 特点：
    // 1. RDD是数据集
    // 2. RDD是编程模型
    // 3. RDD相互之间有依赖关系
    // 4. RDD是可以分区的
    val rdd1: RDD[String] = sc.textFile("dataset/wordcount.txt")

    // 3、处理
    //    1、把整句话拆分为多个单词
    val rdd2: RDD[String] = rdd1.flatMap(item => item.split(" "))
    //    2、把每个单词指定一定词频1
    val rdd3: RDD[(String, Int)] = rdd2.map(item => (item, 1))
    //    3、聚合
    val rdd4: RDD[(String, Int)] = rdd3.reduceByKey((curr, agg) => curr + agg)

    // 4、得到结果
    val result = rdd4.collect()
    result.foreach(item => println(item))
  }

  @Test
  def sparkContext(): Unit = {
    // 1. SparkContext如何编写
    //     1. 创建SparkConf
    val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
    //     2. 创建SparkContext
    val sc = new SparkContext(conf)

    // SparkContext身为大入口API，应该能够创建RDD，并且设置参数，设置JAR包

    // 2. 关闭SparkContext，释放集群资源
  }

  val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
  val sc = new SparkContext(conf)

  // 从本地集合创建
  @Test
  def rddCreationLocal(): Unit = {
    val seq = Seq("Hello1", "Hello2", "Hello3")
    val rdd1: RDD[String] = sc.parallelize(seq, 2)  // 可以不指定分区数
    sc.parallelize(seq)
    val rdd2: RDD[String] = sc.makeRDD(seq, 2)      // 不常见
  }

  // 从文件创建
  @Test
  def rddCreationFiles(): Unit = {
    sc.textFile("file:///...")
    // 1. textFile传入的是什么
    //     * 传入的是一个路径，读取路径
    //     * hdfs://   file://       /.../...（这种方式分为在集群中执行还是在本地执行，集群中读hdfs，本地读文件系统）
    // 2. 是否支持分区
    //     *  传入的path是hdfs:///...    分区是由HDFS中文件的block决定的
    // 3.支持什么平台
    //     * 支持aws和阿里云
  }

  // 从RDD衍生
  @Test
  def rddCreationFromRDD(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3))
    // 通过在rdd上执行算子操作，会生成新的rdd
    // 原地计算
    // str.substr 返回新的字符串，非原地计算
    // 和字符串中的方法很像，字符串不可变，rdd也不可变
    val rdd2: RDD[Int] = rdd1.map(item => item)
  }

  @Test
  def mapTest(): Unit = {
    // 1. 创建RDD
    val rdd1 = sc.parallelize(Seq(1, 2, 3))
    // 2. 执行map操作
    val rdd2 = rdd1.map(item => item * 10)
    // 3. 得到结果
    val result: Array[Int] = rdd2.collect()
    result.foreach(item => println(item))
  }

  @Test
  def flatMapTest(): Unit = {
    // 1. 创建RDD
    val rdd1 = sc.parallelize(Seq("Hello Lily", "Hello Lucy", "Hello Tim"))
    // 2. 处理数据
    val rdd2: RDD[String] = rdd1.flatMap(item => item.split(" "))
    // 3. 得到结果
    val result = rdd2.collect()
    result.foreach(item => println(item))
    // 4. 关闭sc
    sc.stop()
  }

  @Test
  def reduceByKeyTest(): Unit = {
    // 1. 创建RDD
    val rdd1 = sc.parallelize(Seq("Hello Lily", "Hello Lucy", "Hello Tim"))
    // 2. 处理数据
    val rdd2 = rdd1.flatMap(item => item.split(" "))
      .map(item => (item, 1))
      .reduceByKey((curr, agg) => curr + agg)
    // 3. 得到结果
    val result: Array[(String, Int)] = rdd2.collect()
    result.foreach(item => println(item))
    // 4. 关闭sc
    sc.stop()
  }

}



























